﻿using System;
using System.Collections.Generic;
using System.Linq;
using XcoAppSpaces.Contracts.Messaging;
using XcoAppSpaces.Core;
using Microsoft.Ccr.Core;
using System.Threading;
using NUnit.Framework;
using XcoAppSpaces.Contracts.Exceptions;

namespace test.XcoAppSpaces.Core
{
	[TestFixture]
	public class testCommunicationErrorHandling
	{
		[SetUp]
		public void SetUp()
		{
			Dispatcher.ClearCausalities();
		}

		[Test]
		public void TestCommError_PostWithCausality()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				var port = new Port<int>();
				space1.RunWorker(port);

				var remotePort = space2.ConnectWorker<Port<int>>(Environment.MachineName + ":8000");
				
				space1.Dispose();
				Thread.Sleep(100);

				var exceptionPort = new Port<Exception>();
				remotePort.PostWithCausality(1, exceptionPort);
				Assert.IsInstanceOf(typeof(XcoCommunicationException), exceptionPort.TestSync(10000));
				Thread.Sleep(100);
				Assert.That(exceptionPort.ItemCount, Is.EqualTo(0));
				Assert.AreEqual(0, space2.Errors.ItemCount);
			}
		}

		[Test]
		public void TestCommError_Send()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8010"),
				space2 = new XcoAppSpace("tcp.port=8011"))
			{
				var port = new Port<int>();
				space1.RunWorker(port);

				var remotePort = space2.ConnectWorker<Port<int>>(Environment.MachineName + ":8010");

				space1.Dispose();
				Thread.Sleep(100);

				try
				{
					remotePort.Send(1);
					Assert.Fail("XcoCommunicationException should have been thrown!");
				}
				catch (XcoCommunicationException)
				{
				}
				Assert.AreEqual(0, space1.Errors.ItemCount);
				Assert.AreEqual(0, space2.Errors.ItemCount);
			}
		}

		[Test]
		public void TestCommError_Send_CausalityShouldNotActivate()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8020"),
				space2 = new XcoAppSpace("tcp.port=8021"))
			{
				var port = new Port<int>();
				space1.RunWorker(port);

				var remotePort = space2.ConnectWorker<Port<int>>(Environment.MachineName + ":8020");

				space1.Dispose();
				Thread.Sleep(100);
				
				var exceptionPort = new Port<Exception>();
				var c = new Causality("c", exceptionPort);
				Dispatcher.AddCausality(c);
				try
				{
					remotePort.Send(1);
					Assert.Fail("XcoCommunicationException should have been thrown!");
				}
				catch (XcoCommunicationException)
				{
				}
				Dispatcher.RemoveCausality(c);
				Thread.Sleep(100);
				Assert.AreEqual(0, exceptionPort.ItemCount);
				Assert.AreEqual(0, space1.Errors.ItemCount);
				Assert.AreEqual(0, space2.Errors.ItemCount);
			}
		}

		[Test]
		public void TestCommError_Post_Multiple()
		{

			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8030"),
				  space2 = new XcoAppSpace("tcp.port=8031"))
			{
				var port = new Port<int>();
				space1.RunWorker(port);

				var proxy = space2.ConnectWorker<Port<int>>("localhost:8030");
				var exceptionPort = new Port<Exception>();

				space1.Dispose();
				Thread.Sleep(100);

				proxy.PostWithCausality(1, exceptionPort);
				proxy.PostWithCausality(2, exceptionPort);
				Thread.Sleep(5000);
				//SpinWait.SpinUntil(() => exceptionPort.ItemCount >= 2, 10000);

				Assert.AreEqual(2, exceptionPort.ItemCount);
			}
		}

		[Test]
		public void TestAck()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8040"),
				space2 = new XcoAppSpace("tcp.port=8041"))
			{
				space1.RunWorker(new Port<int>());
                var remotePort = space2.Connect<int>(Environment.MachineName + ":8040");

				//install causality with exception port and ack port
				var exceptionPort = new Port<Exception>();
				var ackPort = new Port<Ack>();
				var c = new Causality("c", exceptionPort, ackPort);
				Dispatcher.AddCausality(c);
				
				//successful send
				remotePort.Post(1);
				var ack = ackPort.TestSync(1000);
				Assert.IsNotNull(ack);
				Assert.AreEqual(0, exceptionPort.ItemCount);

				space1.Dispose();
				Thread.Sleep(100);

				//send with communication error
				remotePort.Post(2);
				var ex = exceptionPort.TestSync(15000);
				Assert.IsInstanceOf(typeof(XcoCommunicationException), ex);
				Assert.AreEqual(0, ackPort.ItemCount);

				Dispatcher.RemoveCausality(c);
			}
		}

		[Test]
		public void TestAck_PostWithCausality()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8050"),
							   space2 = new XcoAppSpace("tcp.port=8051"))
			{
				space1.RunWorker(new Port<int>());
				var remotePort = space2.Connect<int>(Environment.MachineName + ":8050");

				//install causality with exception port and ack port
				var exceptionPort = new Port<Exception>();
				var ackPort = new Port<Ack>();

				//successful send
				remotePort.PostWithCausality(1, exceptionPort, ackPort);
				var ack = ackPort.TestSync(1000);
				Assert.IsNotNull(ack);
				Assert.AreEqual(0, exceptionPort.ItemCount);

				space1.Dispose();
				Thread.Sleep(500);

				//send with communication error
				remotePort.PostWithCausality(2, exceptionPort, ackPort);
				var ex = exceptionPort.TestSync(15000);
				Assert.IsInstanceOf(typeof(XcoCommunicationException), ex);
				Assert.AreEqual(0, ackPort.ItemCount);
			}
		}

		[Test]
		public void TestCommError_Send_TcpWithTwoSuccessiveCommunications()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8050"),
							   space2 = new XcoAppSpace("tcp.port=8051"))
			{
				space1.RunWorker(new Port<int>());
				var remotePort = space2.Connect<int>(Environment.MachineName + ":8050");

				//successful send
				remotePort.Send(1);
				space1.Dispose();

				Thread.Sleep(500);
			}

			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8070"),
							   space2 = new XcoAppSpace("tcp.port=8071"))
			{
				space1.RunWorker(new Port<int>());
				var remotePort = space2.Connect<int>(Environment.MachineName + ":8070");

				remotePort.Send(1);
				space1.Dispose();
				Thread.Sleep(500);
				
				//send with communication error
				try
				{
				    remotePort.Send(2);
				    Assert.Fail("exception should have been thrown");
				}
				catch (XcoCommunicationException)
				{
				}
			}
		}

		[Test]
		public void TestCommError_Bidirectional_FirstPostThenSend()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8070"),
							   space2 = new XcoAppSpace("tcp.port=8071"))
			{
				space1.Run<Port<int>>(p =>
				                      	{
											//Console.WriteLine("test");
				                      		p.Post(1);
				                      	});
				var remotePort = space2.Connect<Port<int>>(Environment.MachineName + ":8070");

				var responsePort = new Port<int>();
				for (int i = 0; i < 100; i++ )
					remotePort.Post(responsePort);
				space1.Dispose();
				Thread.Sleep(500);
				Console.WriteLine("retry");

				//send with communication error);
				try
				{
					remotePort.Send(responsePort);
					Assert.Fail("exception should have been thrown");
				}
				catch (XcoCommunicationException ex)
				{
					Console.WriteLine("Comm Ex: " + ex.Message);
				}
			}
		}

		[Test]
		public void TestCommError_SpaceErrorPort()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8060"),
				space2 = new XcoAppSpace("tcp.port=8061"))
			{
				var port = new Port<int>();
				space1.RunWorker(port);

				var remotePort = space2.ConnectWorker<Port<int>>(Environment.MachineName + ":8060");
				Thread.Sleep(100);

				space1.Dispose();
				Thread.Sleep(100);

				remotePort.Post(1);
				var error = space2.Errors.TestSync(10000);
				Assert.IsInstanceOf<XcoCommunicationException>(error.Error);
				Assert.AreEqual(space2.Address, error.SourceAddress);
			}
		}
	}
}
